package com.goxplanet;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Calendar;

public class StockRiskProcessFunction extends CoProcessFunction<StockMarketPrice, StockPosition, StockRisk>{

    private transient ValueState<StockRisk> riskValueState;

    @Override
    public void open(org.apache.flink.configuration.Configuration parameters) {
        ValueStateDescriptor<StockRisk> descriptor =
                new ValueStateDescriptor<>(
                        "risk",
                        StockRisk.class);
        riskValueState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement2(StockPosition position, Context context, Collector<StockRisk> collector) throws Exception {
        StockRisk risk = riskValueState.value();
        if(risk == null){
            risk = new StockRisk(position.ticker, Calendar.getInstance().getTimeInMillis());
        }

        risk.volume = position.volume;
        risk.cost = position.cost;
        if(risk.marketPrice != null) {
            risk.pnl = position.volume * risk.marketPrice - position.cost;
            collector.collect(risk);
        }
        riskValueState.update(risk);
    }

    @Override
    public void processElement1(StockMarketPrice stockMarketPrice, Context context, Collector<StockRisk> collector) throws Exception {
        StockRisk risk = riskValueState.value();
        if(risk == null){
            System.out.println(stockMarketPrice.ticker + " no position data");
            return;
        }
        risk.marketPrice = stockMarketPrice.price;
        risk.pnl = risk.volume * stockMarketPrice.price - risk.cost;
        collector.collect(risk);
        riskValueState.update(risk);
    }
}
